/* 
 * File:   Server.cpp
 * Author: JM, Michał Kołodziejski
 * 
 * Created on 13 maj 2010, 14:35
 */

#include "Server.h"
#include "standardLindaHeaders.h"
#include <signal.h>
#include "TMessage.h"
#include <boost/foreach.hpp>
#include <boost/tokenizer.hpp>
#include <stdexcept>

using boost::lexical_cast;
using boost::bad_lexical_cast;

void printMsg(const boost::shared_ptr<TMessage> msg);
void printClient(const boost::shared_ptr<TClient> client);

int Server::iQueueId = -1;
Server* Server::serverInstance = NULL;

Server::Server() {
    LOG4CXX_INFO(logger, "Creating a server");
    signal(SIGTERM, this->termination_handler);
    signal(SIGINT, this->termination_handler);
    signal(SIGQUIT, this->termination_handler);
}

Server::Server(const Server& orig) {
    // forbidden!
    throw std::exception();
}

Server::~Server() {
}


void Server::init(const int aQueueId){
    this->iQueueId = aQueueId;
}

void Server::do_it() {
    // start timeout-thread
    timeoutThreadFinish = false;
    timeoutThread = boost::shared_ptr<boost::thread>(new boost::thread(boost::bind(&Server::timeoutThreadLoop, this)));

    // process messages
    int res;
    LindaMsg m;
    int msgSize = sizeof(struct LindaMsg) - sizeof(long);
    while (res = msgrcv(iQueueId, &m, msgSize, 0, 0)) {
        LOG4CXX_INFO(logger, "received MSG");
        if (res == -1) {
            std::cout << "Fatal" << std::endl;
            LOG4CXX_INFO(logger, "Error while receiving MSG. Shutdown");
            // clean up and shutdown
            termination_handler(0);
        }

        switch (m.type) {
        case 1:
            {
                putMsg(std::string(m.msg));
                break;
            }
        case 2:
            {
                readMsg(ENormal, m.queueId, m.timeout, std::string(m.msg));
                break;
            }
        case 3:
            {
                readMsg(EExclusive, m.queueId, m.timeout, std::string(m.msg));
                break;
            }
        default:
            //ignore
            break;
        }
    }
}

void Server::putMsg(const std::string aMsg) {
    boost::mutex::scoped_lock lock(mutex);
    LOG4CXX_INFO(logger, "Msg with TXT: "+aMsg);
    boost::shared_ptr<TMessage> msg;
    try {
        msg = parseMsg(aMsg);
        printMsg(msg);
        bool isExclRead = false;

        std::list<boost::shared_ptr<TClient> >::iterator it, it_next;
        for (it = iWaitingClients.begin(); it != iWaitingClients.end(); it = it_next) {
            it_next = it;
            ++it_next;

            if(isMessageFittingClient(msg, *it)){
                //found, so:
                //reply:
                reply((*it)->getQueueId(), 1, msg->toString());
                if ((*it)->getReadType() == EExclusive) {
                    iWaitingClients.erase(it);
                    isExclRead = true;
                    break;
                }
                //else:
                iWaitingClients.erase(it);
            }
        }

        if (!isExclRead)
            iMessages.push_back(msg);
    } catch (MalformedMessageServerException &m) {
        //ignore
        LOG4CXX_INFO(logger, "Malformed msg");
    } catch (std::out_of_range &m) {
        LOG4CXX_INFO(logger, "Malformed msg");
    }
}

boost::shared_ptr<TMessage> Server::parseMsg(const std::string aMsg) const {
    boost::shared_ptr<TMessage> a(new TMessage);
    int len, begin = 0;
    enum tType {
        ENumber, EString
    } tmpType;

    if (aMsg[0] != '(' || aMsg[aMsg.length() - 1] != ')')
        throw MalformedMessageServerException();

    for (int i = 1; i < aMsg.length(); i++) {
        begin = i;

        if (aMsg.at(i) == '"') {
            tmpType = EString; //string
            ++begin;
            ++i;
        } else {
            tmpType = ENumber; //int
        }

        while (i < aMsg.length() &&
                !((tmpType == ENumber && aMsg.at(i) == ',')
                        || (tmpType == ENumber && aMsg.at(i) == ')')
                        || (tmpType == EString && aMsg.at(i) == '"'))) {
            ++i;
        }

        // check if malformed msg
        if(i == aMsg.length()){
            throw MalformedMessageServerException();
        }

        if (aMsg.at(i) == '"') {
            i++;
            len = i - begin - 1;
        } else {
            len = i - begin;
        }

        // add atom
        if (tmpType == ENumber) {
            try {
                a->addAtom(lexical_cast<int> (aMsg.substr(begin, len)));
            } catch (bad_lexical_cast &) {
                throw MalformedMessageServerException();
            }
        } else {
            a->addAtom(aMsg.substr(begin, len));
        }

    }

    return a;
}

void Server::termination_handler(int aSignum) {
    LOG4CXX_INFO(logger, "Got signal");

    // finish timeout thread
    serverInstance->timeoutThreadFinish = true;
    serverInstance->timeoutThread->join();

    //tell all waiting clients to go to hell, because
    //everything has screwed up
    std::list<boost::shared_ptr<TClient> >::iterator it;
    for (it = Server::serverInstance->iWaitingClients.begin(); it != Server::serverInstance->iWaitingClients.end(); ++it) {
        Server::serverInstance->reply((*it)->getQueueId(), 3, "");
    }

    //close the queue:
    if (msgctl(Server::getQueueId(), IPC_RMID, NULL) != 0) {
        //catastrophy
        LOG4CXX_ERROR(logger, "Cannot remove global queue");
    }

    exit(0);
}

void Server::readMsg(const enum TReadType aReadType, const int aPrivQueueId, const int aTimeout,
        const std::string aMessage) {
    boost::mutex::scoped_lock lock(mutex);
    try {
        boost::shared_ptr<std::list<TFilter> > filter = parseFilter(aMessage.substr(1, aMessage.length() - 2));
        boost::shared_ptr<TClient> client = boost::shared_ptr<TClient>(new TClient(aReadType, aPrivQueueId, aTimeout, filter));
        if(aTimeout != -1){
            client->setTimeoutTime(boost::posix_time::second_clock::local_time() + boost::posix_time::seconds(aTimeout));
        }
        bool found = false;

        LOG4CXX_INFO(logger, lexical_cast<std::string>(aPrivQueueId)+" want "+aMessage);

        std::list<boost::shared_ptr<TMessage> >::iterator it, it_next;
        for (it = iMessages.begin(); it != iMessages.end(); it = it_next) {
            it_next = it;
            ++it_next;

            if (isMessageFittingClient(*it, client)) {
                //found, so:
                found = true;
                LOG4CXX_INFO(logger, "Fitting message found - sending response");
                //reply:
                reply(aPrivQueueId, 1, (*it)->toString());
                if (aReadType == EExclusive) {
                    LOG4CXX_INFO(logger, "Removed "+(*it)->toString());
                    iMessages.erase(it);
                }
                break;
            }
        }

        if (!found){
            LOG4CXX_INFO(logger, "No fitting message found - putting client to waiting queue");
            (Server::iWaitingClients).push_back(client);
        }

    } catch (MalformedMessageServerException &m) {
        // send response
        reply(aPrivQueueId, 4, "");
        LOG4CXX_INFO(logger, "Malformed msg");
    }
}

boost::shared_ptr<std::list<TFilter> > Server::parseFilter(const std::string aMessage) const {
    boost::shared_ptr<std::list<TFilter> > tmpList(new std::list<TFilter>());

    std::string partialRegexpString("(\\*|integer|string):(\\*|(==|<|<=|>|>=){0,1}(\"[a-zA-Z0-9]*\")|(==|<|<=|>|>=){0,1}(-{0,1}[0-9]+)),{0,1}");
    boost::regex regexp("("+partialRegexpString+")+");
    boost::cmatch match;
    if(!boost::regex_match(aMessage.c_str(), match, regexp)){
        throw MalformedMessageServerException();
    }

    boost::regex partialRegexp(partialRegexpString);

    boost::match_results<std::string::const_iterator> what;
    boost::match_flag_type flags = boost::match_default;
    std::string::const_iterator start, end;
    start = aMessage.begin();
    end = aMessage.end();
    std::stringstream ss;
    ss << "Whole: " << aMessage << std::endl;
    while(regex_search(start, end, what, partialRegexp, flags)){
        for(int i=0; i<what.size(); ++i){
            ss << i << ": " << std::string(what[i].first, what[i].second) << std::endl;
        }
        ss << std::endl;

        TFieldType fieldType;
        TFilterOperator filterOperator;
        std::string filterInfo("");
        if(what[1] == "*"){
            if(what[2] == "*"){
                fieldType = EAny;
                filterOperator = ANY;
            }
            else if(what[6] != ""){
                // integer
                fieldType = ENumber;
                filterInfo = what[6];
                if(what[5] == "" || what[5] == "=="){
                    filterOperator = EQUAL;
                } else if(what[5] == ">"){
                    filterOperator = GREATER;
                } else if(what[5] == ">="){
                    filterOperator = GREATER_OR_EQUAL;
                } else if(what[5] == "<"){
                    filterOperator = LESS;
                } else if(what[5] == "<="){
                    filterOperator = LESS_OR_EQUAL;
                }
            } else {
                // string
                fieldType = EString;
                filterInfo = what[4];
                filterInfo = filterInfo.substr(1, filterInfo.size()-2); // bez " "
                if(what[3] == "" || what[3] == "=="){
                    filterOperator = EQUAL;
                } else if(what[3] == ">"){
                    filterOperator = GREATER;
                } else if(what[3] == ">="){
                    filterOperator = GREATER_OR_EQUAL;
                } else if(what[3] == "<"){
                    filterOperator = LESS;
                } else if(what[3] == "<="){
                    filterOperator = LESS_OR_EQUAL;
                }
            }


        } else if(what[1] == "integer"){
            fieldType = ENumber;
            if(what[2] == "*"){
                filterOperator = ANY;
            } else {
                filterInfo = what[6];
                if(what[5] == "" || what[5] == "=="){
                    filterOperator = EQUAL;
                } else if(what[5] == ">"){
                    filterOperator = GREATER;
                } else if(what[5] == ">="){
                    filterOperator = GREATER_OR_EQUAL;
                } else if(what[5] == "<"){
                    filterOperator = LESS;
                } else if(what[5] == "<="){
                    filterOperator = LESS_OR_EQUAL;
                }
            }

        } else if(what[1] == "string"){
            fieldType = EString;
            if(what[2] == "*"){
                filterOperator = ANY;
            } else {
                filterInfo = what[4];
                filterInfo = filterInfo.substr(1, filterInfo.size()-2); // bez " "
                if(what[3] == "" || what[3] == "=="){
                    filterOperator = EQUAL;
                } else if(what[3] == ">"){
                    filterOperator = GREATER;
                } else if(what[3] == ">="){
                    filterOperator = GREATER_OR_EQUAL;
                } else if(what[3] == "<"){
                    filterOperator = LESS;
                } else if(what[3] == "<="){
                    filterOperator = LESS_OR_EQUAL;
                }
            }
        }
        tmpList->push_back(TFilter(fieldType, filterOperator, filterInfo));
        start = what[0].second;
    }
    LOG4CXX_INFO(logger, ss.str());
    return tmpList;
}


void Server::reply(const int aPrivQueueId, const int aType, const std::string aMsg) const {
    LindaReplyMsg msg;
    int msgSize = sizeof(struct LindaReplyMsg) - sizeof(long);
    msg.type = aType;
    strncpy(msg.msg, aMsg.c_str(), MAX_QUEUE_MSG_SIZE-1);
    msg.msg[MAX_QUEUE_MSG_SIZE-1] = '\0';
    if (msgsnd(aPrivQueueId, &msg, msgSize, 0) == -1) {
        LOG4CXX_ERROR(logger, "Error while sending "+aMsg+" to:"+lexical_cast<std::string>(aPrivQueueId)+", errno: "
                +lexical_cast<std::string>(errno));
    } else {
        LOG4CXX_INFO(logger, "Sent "+aMsg+" to:"+lexical_cast<std::string>(aPrivQueueId));
    }
}


bool Server::isMessageFittingClient(boost::shared_ptr<TMessage> msg, boost::shared_ptr<TClient> client){
    printClient(client);
    printMsg(msg);
    bool fitting = true;
    boost::shared_ptr<std::list<TFilter> > filters = client->getFilter();
    std::list<TFilter>::iterator it = filters->begin();
    boost::shared_ptr<TAtomBase> atom = msg->getFirstAtom();
    for(; fitting && it != filters->end() && atom.get() != NULL; ++it, atom=atom->getNext()){

        if(it->getType() == EAny && it->getFilterOperator() == ANY){
            // all ok
            continue;
        }

        if(atom->getType() == ENumber && (it->getType() == EAny || it->getType() == ENumber)){
            // integer
            boost::shared_ptr<TAtomNumeric> atomNumeric = boost::dynamic_pointer_cast<TAtomNumeric>(atom);
            int value = -1;
            if(it->getFilterOperator() == ANY){
                continue;
            } else {
                try {
                    value = boost::lexical_cast<int>(it->getFilterInfo());
                } catch(const boost::bad_lexical_cast &e){
                    fitting = false;
                    break;
                }
            }

            switch(it->getFilterOperator()){
            case EQUAL:
                fitting = (atomNumeric->getValue() == value);
                break;
            case GREATER:
                fitting = (atomNumeric->getValue() > value);
                break;
            case GREATER_OR_EQUAL:
                fitting = (atomNumeric->getValue() >= value);
                break;
            case LESS:
                fitting = (atomNumeric->getValue() < value);
                break;
            case LESS_OR_EQUAL:
                fitting = (atomNumeric->getValue() <= value);
                break;
            }


        } else if (atom->getType() == EString && (it->getType() == EAny || it->getType() == EString)){
            // string
            if(it->getFilterOperator() == ANY){
                continue;
            }
            boost::shared_ptr<TAtomString> atomString = boost::dynamic_pointer_cast<TAtomString>(atom);
            int cmpResult = atomString->getValue().compare(it->getFilterInfo());
            switch(it->getFilterOperator()){
            case EQUAL:
                fitting = (cmpResult == 0);
                break;
            case GREATER:
                fitting = (cmpResult > 0);
                break;
            case GREATER_OR_EQUAL:
                fitting = (cmpResult == 0 || cmpResult > 0);
                break;
            case LESS:
                fitting = (cmpResult < 0);
                break;
            case LESS_OR_EQUAL:
                fitting = (cmpResult == 0 || cmpResult < 0);
                break;
            }

        } else {
            fitting = false;
        }
    }

    // unequal number of composites
    if(fitting && ((it != filters->end() && atom.get() == NULL)
            || (it == filters->end() && atom.get() != NULL))){
        fitting = false;
    }

    return fitting;
}



void printClient(const boost::shared_ptr<TClient> client){
    boost::shared_ptr<std::list<TFilter> > filters = client->getFilter();
    std::stringstream ss;
    ss << "Client: " << client->getQueueId() << std::endl;
    for(std::list<TFilter>::iterator it = filters->begin(); it != filters->end(); ++it){
        ss << "  Filter type: " << it->getType();
        ss << ", operator: " << it->getFilterOperator();
        ss << ", data: " << it->getFilterInfo();
        ss << std::endl;
    }
    LOG4CXX_INFO(logger, ss.str());
}


void printMsg(const boost::shared_ptr<TMessage> msg){
    std::stringstream ss;
    ss << "Message: " << std::endl;
    boost::shared_ptr<TAtomBase> atom = msg->getFirstAtom();
    while(atom.get() != NULL){
        if(atom->getType() == ENumber){
            ss << "  Integer: ";
            TAtomNumeric* atomNumeric = (TAtomNumeric*)atom.get();
            ss << atomNumeric->getValue();
        } else if(atom->getType() == EString){
            ss << "  String: ";
            TAtomString* atomString = (TAtomString*)atom.get();
            ss << atomString->getValue();
        }
        ss << std::endl;
        atom = atom->getNext();
    }
    LOG4CXX_INFO(logger, ss.str());
}



void Server::timeoutThreadLoop(){
    while(!timeoutThreadFinish){
        {
            boost::mutex::scoped_lock lock(mutex);
            // check out for timeouts
            boost::posix_time::ptime now = boost::posix_time::second_clock::local_time();
            std::list<boost::shared_ptr<TClient> >::iterator it, it_next;
            for (it = iWaitingClients.begin(); it != iWaitingClients.end(); it = it_next) {
                it_next = it;
                ++it_next;

                if((*it)->getTimeout() != -1 && (*it)->getTimeoutTime() < now){
                    reply((*it)->getQueueId(), 2, "timeout");
                    LOG4CXX_INFO(logger, "TIMEOUT - erasing client");
                    iWaitingClients.erase(it);
                }

            }

        }
        sleep(2);
    }
    LOG4CXX_INFO(logger, "thread finish");
}
